package service

import (
	"encoding/json"
	"github.com/gofrs/uuid"
	"github.com/mohae/deepcopy"
	"strconv"
	"strings"
	"sync"
	"time"
	"vrcm/src/vrcm/app/admin/models"
	"vrcm/src/vrcm/app/admin/models/live_schedule"
	baiduService "vrcm/src/vrcm/app/admin/service/baidu"
	baiduHubDto "vrcm/src/vrcm/app/admin/service/dto/baidu"
	liveDto "vrcm/src/vrcm/app/admin/service/dto/live"
	dto "vrcm/src/vrcm/app/admin/service/dto/live_schedule"
	StreamHubDto "vrcm/src/vrcm/app/admin/service/dto/stream_hub"
	liveService "vrcm/src/vrcm/app/admin/service/live"
	streamHubService "vrcm/src/vrcm/app/admin/service/stream_hub"
	"vrcm/src/vrcm/common/service"
	ext "vrcm/src/vrcm/envconfig"
)

type AuditTaskService struct {
	service.Service
}

func (e *AuditTaskService) DeleteAuditTask(wg *sync.WaitGroup, workTotalNum int, partNum *int, tasks []live_schedule.AuditTask, sourcesMap map[int]live_schedule.SourceBasic, basic live_schedule.ChannelBasic, liveService *liveService.LiveBasic, streamHubService *streamHubService.StreamHubService, liveBasicService SourceBasicService) error {

	var m sync.Mutex
	baiduService := &baiduService.BaiduService{}
	baiduService.Log = e.Log
	liveBasicService.Orm = e.Orm
	//// 创建任务通道
	//taskCh := make(chan live_schedule.AuditTask, 3)
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.AuditTask, wg *sync.WaitGroup) {
			//v := v
			//source := sourcesMap[v.SourceId]
			////删除自定义
			//if source.Kind == 2 && len(source.TaskId) > 0 {
			//	err := liveBasicService.DeleteCustomSource(source)
			//	if err != nil {
			//		e.Log.Errorf("DeleteCustomSource error:%s \r\n", err)
			//	}
			//}
			defer wg.Done()
			//删除协转任务流
			if len(v.TaskId) > 0 {
				err := streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
				}
				param := make(map[string]interface{})
				param["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0", v.TaskId).Updates(&param).Error
				if err != nil {
					e.Log.Errorf("updateScheduleRecord error:%s \r\n", err)
				}
			}
			//删除协转任务流
			if len(v.PullTaskId) > 0 {
				err := streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
				}
				param := make(map[string]interface{})
				param["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0", v.PullTaskId).Updates(&param).Error
				if err != nil {
					e.Log.Errorf("updateScheduleRecord error:%s \r\n", err)
				}
			}

			//删除协转任务流
			if len(v.LiveId) > 0 {
				err := liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: v.TenantId, LiveId: v.PullTaskId})
				if err != nil {
					e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
				}
			}

			//停止审核任务
			if len(v.AuditTaskId) > 0 {
				err := baiduService.StopAudit(&baiduHubDto.StopAuditReq{TaskId: v.AuditTaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("StopAudit error:%s \r\n", err)
				}
			}
			err := e.Orm.Delete(&v).Error
			if err != nil {
				e.Log.Errorf("SaveWorkflow_Delete error:%s \r\n", err)
			}

			m.Lock()
			*partNum++
			err = e.CalculateProgress(*partNum, workTotalNum, basic)
			if err != nil {
				e.Log.Errorf("CalculateProgress error:%s \r\n", err)
			}
			m.Unlock()
		}(tasks[i], wg)
	}
	return nil

}

func (e *AuditTaskService) SaveAuditTask(wg *sync.WaitGroup, workTotalNum int, partNum *int, tasks []live_schedule.AuditTask, upMap map[int]live_schedule.TrafficOn, offMap map[int]live_schedule.TrafficOff, basic live_schedule.ChannelBasic, mediaProcess live_schedule.MediaProcess, sourcesMap map[int]live_schedule.SourceBasic, auditMap map[int]live_schedule.AuditTask) error {

	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	baiduService := &baiduService.BaiduService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	baiduService.Log = e.Log
	staticsService := &ChannelStatisticsService{}
	staticsService.Orm = e.Orm
	staticsService.Log = e.Log
	now := time.Now().Unix()
	//var m sync.Mutex

	originTasks := make([]live_schedule.AuditTask, 0, 5)
	autoTasks := make([]live_schedule.AuditTask, 0, 5)
	//taskCh := make(chan live_schedule.AuditTask, 3)
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.AuditTask, wg *sync.WaitGroup) {
			defer wg.Done()
			v1 := live_schedule.AuditTask{}
			//获取上车点
			up := upMap[v.TrafficOnId]

			//获取下车点配置
			off, ok := offMap[v.TrafficOffId]
			if ok {
				//信源
				source, ok := sourcesMap[v.SourceId]
				if !ok {
					e.Log.Errorf("信源已被删除")
					return
				}
				e.Log.Trace("当前使用信源：", source.SourceName)
				pullUrlPart := ""
				pullDomain := ""
				if (len(v.PushUrl) <= 0 && off.StreamType == 1) || (off.IsCustom == 1 && len(v.CustomOutUrl) <= 0) {
					e.Log.Trace("不存在转推地址或者拉流地址")
				} else {
					//输出参数
					target := StreamHubDto.Target{}
					target.Format = protocols[off.TrafficOffType]
					target.Id = v.OutNodeId
					//输入参数
					sourceParam := StreamHubDto.Source{}

					//使用转码源还是原始源
					if up.StreamType == 1 {
						e.Log.Trace("当前使用原始源：", up.SourceId)
						sourceParam.Id = v.InNodeId
						sourceParam.Url = source.OutUdpUrl
					} else {
						sourceParam.Id = v.InNodeId
						sourceParam.Url = up.TransUdpUrl
					}

					req := StreamHubDto.StreamHubReq{}

					if len(v.TaskId) > 0 {
						req.Id = v.TaskId
					}

					req.Name = ""
					req.Type = "live"
					req.TenantId = basic.TenantId
					req.Description = ""
					req.Input = StreamHubDto.Input{
						Format: protocols[5],
						Active: 0,
						Policy: StreamHubDto.Policy{
							Mode: "manual",
						},
						Type:    "push",
						Sources: []StreamHubDto.Source{sourceParam},
					}

					//第三方拉流+内网 输出类型为pull 其他情况为push
					if off.StreamType == 2 && off.Area == 2 {
						target.Type = "pull"
					} else {
						target.Type = "push"
					}
					delay := StreamHubDto.Delay{
						Enable:   "no",
						Duration: 10,
					}
					//if mediaProcess.PlayType == 2 {
					//	delay.Enable = "yes"
					//	delay.Duration = mediaProcess.Delay * 1000
					//}

					req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/audit-task-call-back"
					hub := &StreamHubDto.StreamHubRes{}
					target1 := deepcopy.Copy(target).(StreamHubDto.Target)
					target1.Type = "pull"
					//外网+第三方拉流需要先调用全球直播
					if (off.Area == 3 || off.Area == 1) && off.StreamType == 2 {
						target.Format = "rtmp"
						streamReq := e.GenerateInsertLiveParam(basic.TenantId)
						res, err := liveService.CreateLiveStream(streamReq)
						if err != nil {
							e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
							//tx.Rollback()
							//return err
						}

						v.LiveId = res.LiveId
						//记录额外信息
						extraMap := make(map[string]string)
						extraMap["rtmp"] = res.RtmpUrl
						extraMap["hls"] = res.HlsUrl
						extraMap["flv"] = res.FlvUrl

						data, err := json.Marshal(extraMap)
						if err != nil {
							e.Log.Errorf("Marshal error:%s \r\n", err)
							//tx.Rollback()
							//return err
						}
						v.ExtraInfo = string(data)
						pushes := strings.Split(res.PushUrl, models.Delimiter)
						if len(pushes) > 0 {
							target.Url = pushes[0]
							v.PushUrl = pushes[0]
						}

						v.OutUrl = extraMap[protocols[off.TrafficOffType]]

						parts := strings.Split(v.OutUrl, "/")
						liveStream := parts[len(parts)-1]
						queryString := strings.Split(liveStream, ".")[0]
						pullUrlPart = queryString
						//如果是外网并且拉流地址
						if off.Area == 3 && off.StreamType == 2 {
							v.Area = 1
						}

					} else if off.StreamType == 1 {
						target.Url = v.PushUrl
					}
					target.RedefinePlayUrlEnable = "2"
					params := StreamHubDto.RedefinePlayUrlParams{}
					target.RedefinePlayUrlParams = params
					req1 := deepcopy.Copy(req).(StreamHubDto.StreamHubReq)
					//如果是自定义拉流地址
					if off.StreamType == 2 && off.TrafficOffType == 2 && (off.Area == 2 || off.Area == 3) && off.IsCustom == 1 {
						arr := strings.Split(v.CustomOutUrl, "/")
						if off.Area == 3 {
							if len(arr) > 1 {
								target1.RedefinePlayUrlEnable = "1"
								params.Path = arr[0]
								index := strings.Index(arr[1], ".")
								params.Sub = arr[1][0:index]
								target1.RedefinePlayUrlParams = params
							}
						} else {
							if len(arr) > 1 {
								target.RedefinePlayUrlEnable = "1"
								params.Path = arr[0]
								index := strings.Index(arr[1], ".")
								params.Sub = arr[1][0:index]
								target.RedefinePlayUrlParams = params
							}
						}

					}
					//创建协助任务
					req.Output = StreamHubDto.Output{
						Targets: []StreamHubDto.Target{target},
						Delay:   delay,
						Acl:     StreamHubDto.Acl{},
					}
					req1.Output = StreamHubDto.Output{
						Targets: []StreamHubDto.Target{target1},
						Delay:   delay,
						Acl:     StreamHubDto.Acl{},
					}
					hub, err := streamHubService.CreateStreamHub(&req)
					if err != nil {
						e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}

					if hub != nil {
						if len(hub.Input.Sources) > 0 {
							v.InUrl = hub.Input.Sources[0].UdpAddress
							v.InNodeId = hub.Input.Sources[0].Id
						}
						//如果是外网拉流就不需要再次接收返回的输出url
						if len(hub.Output.Targets) > 0 {

							if !((off.Area == 1 || off.Area == 3) && off.StreamType == 2) {
								v.OutUrl = hub.Output.Targets[0].Url
							}
							v.OutNodeId = hub.Output.Targets[0].Id
							if hub.Output.Targets[0].RedefinePlayUrlEnable == "1" {
								path := hub.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
								v.CustomOutUrl = path
							}
						}
						v.StateUpdateTime = now
						v.TaskState = 3
						v.PullStateUpdateTime = now
						v.PullTaskState = 3
						v.TaskId = hub.TaskId
						v.OutputStreamId = hub.OutputStreamName
						v.UpdateTime = now
					}

					//自动调度内网情况
					if off.Area == 3 && off.StreamType == 2 && (off.TrafficOffType == 1 || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
						req1.OutputStreamName = pullUrlPart
						hub1, err := streamHubService.CreateStreamHub(&req1)
						if err != nil {
							e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
							//tx.Rollback()
							//return err
						}
						if hub1 != nil {
							v1 = deepcopy.Copy(v).(live_schedule.AuditTask)
							if len(hub1.Input.Sources) > 0 {
								v1.InUrl = hub1.Input.Sources[0].UdpAddress
								v1.InNodeId = hub1.Input.Sources[0].Id
							}

							//如果是外网拉流就不需要再次接收返回的输出url
							if len(hub1.Output.Targets) > 0 {
								v1.OutUrl = hub1.Output.Targets[0].Url
								pullDomain = v1.OutUrl[0:strings.LastIndex(v1.OutUrl, "/")]
								v1.OutNodeId = hub1.Output.Targets[0].Id
								if hub1.Output.Targets[0].RedefinePlayUrlEnable == "1" {
									path := hub1.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub1.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
									v1.CustomOutUrl = path
								}
							}
							v.OutUrl = pullDomain + "/" + v.OutUrl[strings.LastIndex(v.OutUrl, "/")+1:]
							v1.StateUpdateTime = time.Now().Unix()
							v1.TaskState = 3
							v1.TaskId = hub1.TaskId
							v1.Area = 2
							v1.OutputStreamId = hub1.OutputStreamName
							v1.UpdateTime = time.Now().Unix()
							v1.PullStateUpdateTime = now
							v1.PullTaskState = 3
						}
					}

					//如果通道状态是启用 就开启协助任务
					e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Find(&basic)
					if basic.ChannelState == 1 {
						if source.Kind == 2 && len(source.TaskId) > 0 {
							err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: source.TaskId})
							if err != nil {
								e.Log.Errorf("StartStreamHub error:%s \r\n", err)
								//tx.Rollback()
								//return err
							}
						}
						err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: hub.TaskId})
						if err != nil {
							e.Log.Errorf("StartStreamHub error:%s \r\n", err)
							//tx.Rollback()
							//return err
						}
						if len(v1.TaskId) > 0 {
							err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: v1.TaskId})
							if err != nil {
								e.Log.Errorf("StopStreamHub error:%s \r\n", err)
								//tx.Rollback()
								//return err
							}
						}

						//记录任务启动记录用于统计
						err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, v.TaskId, 1, now)
						if err != nil {
							e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
						}
					}
				}

				//默认设置成审核通过
				v.IsAudit = 1
				v1.IsAudit = 1
				audit := auditMap[v.SourceId]

				//获取
				if len(v.PullTaskId) <= 0 || len(v1.PullTaskId) <= 0 && audit.Id > 0 {

					//获取
					req := StreamHubDto.StreamHubReq{}
					//输出参数
					target2 := StreamHubDto.Target{}
					target2.Format = protocols[1]

					//输入参数
					sourceParam1 := StreamHubDto.Source{}
					sourceParam1.Url = source.OutUdpUrl
					req.Name = ""
					req.Type = "live"
					req.TenantId = v.TenantId
					req.Description = ""
					req.Input = StreamHubDto.Input{
						Format: protocols[5],
						Active: 0,
						Policy: StreamHubDto.Policy{
							Mode: "manual",
						},
						Type:    "push",
						Sources: []StreamHubDto.Source{sourceParam1},
					}

					if ext.ExtConfig.ServiceConfig.Area == "1" || ext.ExtConfig.ServiceConfig.Area == "2" || ext.ExtConfig.ServiceConfig.Area == "" {
						target2.Format = "rtmp"
						target2.Type = "push"
						streamReq := e.GenerateInsertLiveParam(v.TenantId)
						res, err := liveService.CreateLiveStream(streamReq)
						if err != nil {
							e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
							//tx.Rollback()
							//return err
						}
						//记录额外信息
						extraMap := make(map[string]string)
						extraMap["rtmp"] = res.RtmpUrl
						extraMap["hls"] = res.HlsUrl
						extraMap["flv"] = res.FlvUrl

						pushes := strings.Split(res.PushUrl, models.Delimiter)
						if len(pushes) > 0 {
							target2.Url = pushes[0]
							v.PushUrl = pushes[0]
						}
						//*(&tasks[k].OutUrl) = extraMap["rtmp"]
						v.AuditPullUrl = extraMap["rtmp"]
						v1.AuditPullUrl = extraMap["rtmp"]
					} else {
						target2.Format = "pull"
					}

					delay1 := StreamHubDto.Delay{
						Enable:   "no",
						Duration: 10,
					}
					//if mediaProcess.PlayType == 2 {
					//	delay1.Enable = "yes"
					//	delay1.Duration = mediaProcess.Delay * 1000
					//}

					//创建协助任务
					req.Output = StreamHubDto.Output{
						Targets: []StreamHubDto.Target{target2},
						Delay:   delay1,
						Acl:     StreamHubDto.Acl{},
					}
					hub2, err := streamHubService.CreateStreamHub(&req)
					if err != nil {
						e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}

					if hub2 != nil {
						if len(hub2.Input.Sources) > 0 {
							v.InUrl = hub2.Input.Sources[0].UdpAddress
							v1.InUrl = hub2.Input.Sources[0].UdpAddress
						}
						//如果是外网拉流就不需要再次接收返回的输出url
						if len(hub2.Output.Targets) > 0 {
							if ext.ExtConfig.ServiceConfig.Area == "3" {
								v.OutUrl = hub2.Output.Targets[0].Url
								v1.OutUrl = hub2.Output.Targets[0].Url
							}
						}
						v.PullTaskId = hub2.TaskId
						v1.PullTaskId = hub2.TaskId
					}
					if basic.ChannelState == 1 {
						err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: v.PullTaskId})
						if err != nil {
							e.Log.Errorf("StartStreamHub error:%s \r\n", err)
							//tx.Rollback()
							//return err
						}
						if len(v1.PullTaskId) > 0 {
							err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: v1.PullTaskId})
							if err != nil {
								e.Log.Errorf("StartStreamHub error:%s \r\n", err)
								//tx.Rollback()
								//return err
							}
							////记录任务启动记录用于统计
							//err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, v.PullTaskId, 4, now)
							//if err != nil {
							//	e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
							//}
						}

						//记录任务启动记录用于统计
						err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, v.TaskId, 1, now)
						if err != nil {
							e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
						}
					}
				}

				//没有审核任务且调度任务开启 需要创建审核任务
				if len(v.AuditTaskId) <= 0 && basic.ChannelState != 2 && audit.Id <= 0 && mediaProcess.EnableAi == 1 {
					auditId, endTime, err := e.CreateAuditTask(v.OutUrl, mediaProcess, baiduService)
					if err != nil {
						e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
					v.AuditTaskId = auditId
					v1.AuditTaskId = auditId
					v.AuditEndTime = endTime
					v1.AuditEndTime = endTime
					v.IsAudit = 1
					v1.IsAudit = 1
					//记录任务启动记录用于统计
					err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, auditId, 4, now)
					if err != nil {
						e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
					}
					err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, auditId, 4, now)
					if err != nil {
						e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
					}
				}
				if audit.Id > 0 {
					v.IsAudit = audit.IsAudit
					v1.IsAudit = audit.IsAudit
				}

				err := e.Orm.Save(&v).Error
				if err != nil {
					e.Log.Errorf("Save AuditTask error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				originTasks = append(originTasks, v)
				if len(v1.TaskId) > 0 {
					err := e.Orm.Save(&v1).Error
					if err != nil {
						e.Log.Errorf("Save AuditTask error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
					autoTasks = append(autoTasks, v1)
				}
			} else {
				err := e.Orm.Save(&v).Error
				if err != nil {
					e.Log.Errorf("Save AuditTask error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				originTasks = append(originTasks, v)
			}

			//设置黑白名单
			channelTaskService := ChannelTaskService{}
			channelTaskService.Log = e.Log
			taskIds := make([]string, 0, 5)
			taskIds = append(taskIds, v.OutNodeId)
			taskIds = append(taskIds, v1.OutNodeId)
			channelTaskService.ProcessBlackWhite(taskIds, off)
			//m.Lock()
			//*partNum++
			//err := e.CalculateProgress(*partNum, workTotalNum, basic)
			//if err != nil {
			//	e.Log.Errorf("CalculateProgress error:%s \r\n", err)
			//}
			//m.Unlock()
		}(tasks[i], wg)
	}
	return nil
}

func (e *AuditTaskService) GenerateInsertLiveParam(tenantId string) *liveDto.LiveStreamInsertReq {
	streamReq := &liveDto.LiveStreamInsertReq{}
	streamReq.EventName = ""
	streamReq.TenantId = tenantId
	streamReq.ExpireTime = 10 * 365 * 24 * 60 * 60
	streamReq.IMEnable = 0
	streamReq.WebPusherEnable = 0
	streamReq.PullOrPush = ""
	streamReq.AutoRecord = "0"
	streamReq.EventType = "1"
	streamReq.ServiceArea = "cn_mainland"
	return streamReq
}

func (e *AuditTaskService) CreateAuditTask(url string, process live_schedule.MediaProcess, baiduService *baiduService.BaiduService) (string, int64, error) {
	now := time.Now()
	res := &baiduHubDto.AuditRes{}
	auditReq := &baiduHubDto.AuditReq{}
	auditReq.TenantId = process.TenantId
	//auditReq.StrategyId = int64(ext.ExtConfig.ServiceConfig.StrategyId)
	auditReq.StrategyId = 27127
	auditReq.StreamUrl = url
	auditReq.StreamType = "rtmp"
	u, _ := uuid.NewV4()
	auditReq.ExtId = u.String()
	auditReq.StartTime = now.Add(time.Second * 10).UnixMilli()
	auditReq.EndTime = now.Add(time.Hour * 24).UnixMilli()
	auditReq.StreamName = "global_live"
	auditReq.StreamDealInfo.VideoFrequency = 5
	auditReq.SubEvents.VideoSubEvents = "2,3,4"
	auditReq.SubEvents.AudioSubEvents = "2,3,4"
	auditReq.DetectType = 1
	auditReq.Region = "hb"
	//auditReq.NoticeUrl = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/audit-call-back"
	auditReq.NoticeUrl = "http://vpn.videosolar.com:18000" + "/api/v1/channel-task/audit-call-back"
	res, err := baiduService.AuditSubmit(auditReq)
	if err != nil {
		e.Log.Errorf("AuditSubmit error:%s \r\n", err)
		return "", 0, err
	}

	return res.Data.TaskId, now.Add(time.Hour * 24).Unix(), nil
}

func (e *AuditTaskService) StartAuditTasks(wg *sync.WaitGroup, tasks []live_schedule.AuditTask, m live_schedule.MediaProcess, sourceAuditMap map[int]struct{}, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.AuditTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	baiduService := &baiduService.BaiduService{}
	baiduService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	auditTaskService := AuditTaskService{}
	auditTaskService.Log = e.Log
	auditTaskService.Orm = e.Orm
	staticsService := &ChannelStatisticsService{}
	staticsService.Orm = e.Orm
	staticsService.Log = e.Log

	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()
	//var mu sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.AuditTask, wg *sync.WaitGroup) {
			defer wg.Done()

			//如果是审核未通过的任务不能被开启
			if v.IsAudit == 2 {
				return
			}
			//source := sourceMap[v.SourceId]
			//if source.Kind == 2 {
			//	//开启自定义信源
			//	err := sourceBasicService.StartSource(&dto.StartSourceReq{TenantId: v.TenantId, Id: source.Id, TaskId: source.TaskId})
			//	if err != nil {
			//		e.Log.Errorf("StartStreamHub error:%s \r\n", err)
			//		//tx.Rollback()
			//		//return err
			//	}
			//}
			if len(v.TaskId) > 0 {
				//开启协助任务
				err := streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				//创建调度统计记录
				err = staticsService.CreateScheduleRecord(v.TenantId, v.ChannelId, v.TaskId, 1, now)
				if err != nil {
					e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
				}
			}
			if len(v.PullTaskId) > 0 {
				//开启协助任务
				err := streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.PullTaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
			}
			if len(v.AuditTaskId) > 0 {
				//停止审核
				err := baiduService.StopAudit(&baiduHubDto.StopAuditReq{TaskId: v.AuditTaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("StopAudit error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				//更新审核任务记录结束时间
				param := make(map[string]interface{})
				param["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =4", v.AuditTaskId).Updates(&param).Error
				if err != nil {
					e.Log.Errorf("ProcessWorkflow Updates ScheduleRecord error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
			}
			//重新添加审核任务
			_, ok := sourceAuditMap[v.SourceId]
			if ((len(v.AuditTaskId) <= 0 && !ok) || (len(v.AuditTaskId) > 0 && ok)) && m.EnableAi == 1 {
				auditId, endTime, err := auditTaskService.CreateAuditTask(v.AuditPullUrl, m, baiduService)
				if err != nil {
					e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				if len(auditId) > 0 {
					param := make(map[string]interface{})
					param["audit_task_id"] = auditId
					param["audit_end_time"] = endTime
					err = e.Orm.Model(&live_schedule.AuditTask{}).Where("id=?", v.Id).Updates(param).Error
					if err != nil {
						e.Log.Errorf("update AuditTask error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
					sourceAuditMap[v.SourceId] = struct{}{}

					//审核统计记录
					err = staticsService.CreateScheduleRecord(v.TenantId, v.ChannelId, auditId, 4, now)
					if err != nil {
						e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
					}
				}
			}
		}(tasks[i], wg)
		//mu.Lock()
		//*partNum++
		//result := float64(*partNum) / float64(workNum)
		//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
		//param["progress"] = formattedResult
		//if workNum >= *partNum {
		//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", m.ChannelId).Updates(param).Error
		//	if err != nil {
		//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
		//	}
		//} else {
		//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", m.ChannelId).Update("progress", formattedResult).Error
		//	if err != nil {
		//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
		//	}
		//}
		//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
		//mu.Unlock()
	}

	return nil

}

func (e *AuditTaskService) StopAuditTasks(wg *sync.WaitGroup, tasks []live_schedule.AuditTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.AuditTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	baiduService := &baiduService.BaiduService{}
	baiduService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	var mu sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.AuditTask, wg *sync.WaitGroup) {
			defer wg.Done()

			//up := sourceMap[v.SourceId]
			//if up.Kind == 2 {
			//	err := sourceBasicService.StopSource(&dto.StopSourceReq{TenantId: v.TenantId, TaskId: up.TaskId})
			//	if err != nil {
			//		e.Log.Errorf("StopSource error:%s \r\n", err)
			//	}
			//}
			//停止协助任务
			if len(v.TaskId) > 0 {
				err := streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{Id: v.TaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("ProcessWorkflow StopStreamHub error:%s \r\n", err)

				}
				//更新任务记录结束时间
				param2 := make(map[string]interface{})
				param2["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =1", v.TaskId).Updates(&param2).Error
				if err != nil {
					e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
				}
			}
			if len(v.PullTaskId) > 0 {
				err := streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{Id: v.PullTaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("ProcessWorkflow StopStreamHub error:%s \r\n", err)

				}
				//scheduleIds = append(scheduleIds, v.PullTaskId)
			}
			if len(v.AuditTaskId) > 0 {
				err := baiduService.StopAudit(&baiduHubDto.StopAuditReq{TenantId: v.TenantId, TaskId: v.AuditTaskId})
				if err != nil {
					e.Log.Errorf("ProcessWorkflow StopStreamHub error:%s \r\n", err)
				}
				//更新审核任务记录结束时间
				param2 := make(map[string]interface{})
				param2["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and  type =4", v.AuditTaskId).Updates(&param2).Error
				if err != nil {
					e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
				}
			}

			mu.Lock()
			*partNum++
			result := float64(*partNum) / float64(workNum)
			formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
			param["progress"] = formattedResult
			if *partNum == workNum {
				err := e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
				if err != nil {
					e.Log.Errorf("Update ChannelTask error:%s \r\n", err)
				}
			} else {
				err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
				if err != nil {
					e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
				}
			}
			e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
			mu.Unlock()
		}(tasks[i], wg)
	}

	return nil

}

func (e *AuditTaskService) AuditTaskCallBack(c *dto.TargetCallBackReq) error {

	//更新通道状态
	param := make(map[string]interface{})
	if c.TransferState == "Capturing" {
		param["task_state"] = 1
	} else {
		param["task_state"] = 2
	}
	param["state_update_time"] = c.TransferStateLastUpdateTime
	err := e.Orm.Model(live_schedule.AuditTask{}).Where("task_id=? and out_node_id=?", c.Id, c.TaskId).Updates(param).Error
	if err != nil {
		e.Log.Errorf("AuditTaskCallBack error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *AuditTaskService) AuditResultCallBack(c *dto.AuditResult) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service CreateCortexTask error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		return err
	}

	if len(c.VideoFrameList) <= 0 {
		return nil
	}
	e.Log.Trace("审核结果信息为：", *c)
	streamHubService := &streamHubService.StreamHubService{}
	baiduService := &baiduService.BaiduService{}
	streamHubService.Log = e.Log
	baiduService.Log = e.Log
	//审核任务
	auditTask := live_schedule.AuditTask{}
	err := tx.Model(&live_schedule.AuditTask{}).Where("audit_task_id=?", c.TaskId).Find(&auditTask).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get AuditTask error:%s \r\n", err)
		return err
	}
	if auditTask.Id <= 0 {
		tx.Commit()
		return err
	}

	auditTasks := make([]live_schedule.AuditTask, 0, 5)
	err = tx.Model(&live_schedule.AuditTask{}).Where("source_id=?", auditTask.SourceId).Find(&auditTasks).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get AuditTasks error:%s \r\n", err)
		return err
	}
	//调度任务
	channelBasic := live_schedule.ChannelBasic{}
	err = tx.Model(&live_schedule.ChannelBasic{}).Where("id=?", auditTask.ChannelId).Find(&channelBasic).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
	}

	//获取审核节点信息
	var media live_schedule.MediaProcess
	err = tx.Model(&live_schedule.MediaProcess{}).Where("channel_id=?", auditTask.ChannelId).Find(&media).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}

	scheduleIds := make([]string, 0, 5)
	auditScheduleIds := make([]string, 0, 5)
	//启动ai审核之后 需要设置审核记录
	if media.EnableAi == 1 {
		err := tx.Model(&live_schedule.SourceBasic{}).
			Where("id=?", auditTask.SourceId).
			Update("audit_state", 2).
			Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("UpdateAuditState error:%s \r\n", err)
			return err
		}
		now := time.Now().Unix()
		history := live_schedule.AuditHistory{}
		history.SourceID = auditTask.SourceId
		history.AuditState = 2
		history.TenantID = auditTask.TenantId
		history.Account = "智能审核"
		history.AuditTimeStamp = now
		history.CreateTime = now
		history.UpdateTime = now
		err = tx.Model(&live_schedule.AuditHistory{}).Create(&history).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("create AuditHistory error:%s \r\n", err)
			return err
		}

		for _, v := range auditTasks {
			//停止任务
			if len(v.TaskId) > 0 {
				err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("StopStreamHub error:%s \r\n", err)
					return err
				}
				scheduleIds = append(scheduleIds, v.TaskId)
			}
			if len(v.PullTaskId) > 0 {
				err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.PullTaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("StopStreamHub error:%s \r\n", err)
					return err
				}
			}
			if len(v.AuditTaskId) > 0 {
				err = baiduService.StopAudit(&baiduHubDto.StopAuditReq{TaskId: v.AuditTaskId, TenantId: v.TenantId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("StopAuditReq error:%s \r\n", err)
					return err
				}
				auditScheduleIds = append(auditScheduleIds, v.AuditTaskId)
			}
		}

		//更新审核任务状态不通过
		param := make(map[string]interface{})
		param["is_audit"] = 2
		param["is_enable"] = 2
		err = tx.Model(live_schedule.AuditTask{}).Where("source_id = ?", auditTask.SourceId).Updates(&param).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("update AuditTask error:%s \r\n", err)
			return err
		}

		//更新调度任务结束时间
		if len(scheduleIds) > 0 {
			param1 := make(map[string]interface{})
			param1["end_time"] = now
			err := tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0 and  type =1", scheduleIds).Updates(param1).Error
			if err != nil {
				e.Log.Errorf("ChannelTaskService updateScheduleRecord error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

		if len(auditScheduleIds) > 0 {
			//更新审核任务记录结束时间
			param2 := make(map[string]interface{})
			param2["end_time"] = now
			err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0 and  type =4", auditScheduleIds).Updates(&param2).Error
			if err != nil {
				e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
	}

	now := time.Now().Unix()
	results := make([]live_schedule.AuditResult, 0, 5)

	for _, v := range c.VideoFrameList {
		v := v
		flagMap := map[int]struct{}{}
		for _, d := range v.FrameAuditResult.Data {
			violationType := 0
			//if strings.Contains(d.Type, "sexual") {
			//	violationType = 1
			//} else if strings.Contains(v.Type, "terror") {
			//	violationType = 2
			//} else if strings.Contains(v.Type, "disgust") {
			//	violationType = 3
			//} else if strings.Contains(v.Type, "ad_") {
			//	violationType = 4
			//} else if strings.Contains(v.Type, "politic") {
			//	violationType = 5
			//} else {
			//	violationType = 6
			//}
			//_, ok := flagMap[violationType]
			//if ok {
			//	continue
			//}
			switch d.Type {
			case 1:
				violationType = 1
			case 2:
				violationType = 2
			case 3:
				violationType = 3
			case 4:
				violationType = 4
			case 5:
				violationType = 5
			default:
				violationType = 6

			}
			flagMap[violationType] = struct{}{}
			result := live_schedule.AuditResult{}
			result.TenantId = auditTask.TenantId
			result.ChannelId = auditTask.ChannelId
			result.TaskId = auditTask.TaskId
			result.AuditTaskId = c.TaskId
			result.ViolationType = violationType
			result.SourceId = auditTask.SourceId
			result.FrameTimeStamp = v.FrameTimestamp
			result.LogId = v.FrameAuditResult.LogId
			result.SubType = d.SubType
			result.ImgUrl = v.FrameUrl
			result.CreateTime = now
			result.UpdateTime = now
			result.UpdateTime = now
			result.AuditType = 1
			results = append(results, result)
		}

	}
	if len(results) > 0 {
		err = tx.Model(&live_schedule.AuditResult{}).Create(&results).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("create AuditResult error:%s \r\n", err)
			return err
		}
	}

	tx.Commit()
	return nil
}

func (e *AuditTaskService) AuditStatisticResult(c *dto.AuditStatisticReq) (*[]live_schedule.AuditResult, error) {

	results := make([]live_schedule.AuditResult, 0, 5)
	err := e.Orm.Model(&live_schedule.AuditResult{}).Where("tenant_id=? and channel_id=? and violation_type =? and audit_type =1", c.TenantId, c.ChannelId, c.Type).Find(&results).Error
	if err != nil {
		e.Log.Errorf("create AuditResult error:%s \r\n", err)
		return nil, err
	}
	return &results, nil
}

func (e *AuditTaskService) TimeAuditTask() {
	baiduService := &baiduService.BaiduService{}
	baiduService.Log = e.Log

	auditTaskService := AuditTaskService{}
	auditTaskService.Log = e.Log
	auditTaskService.Orm = e.Orm
	//获取需要定时启动的审核通道
	tasks := make([]live_schedule.AuditTask, 0, 5)
	err := e.Orm.Model(live_schedule.AuditTask{}).Where("is_audit=1").Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
	}
	now := time.Now().Unix()

	channelIds := make([]int, 0, 5)
	for _, v := range tasks {
		channelIds = append(channelIds, v.ChannelId)
	}

	//获取媒体处理节点
	medias := make([]live_schedule.MediaProcess, 0, 5)
	err = e.Orm.Model(&live_schedule.MediaProcess{}).Where("channel_id in ? and media_type = 3", channelIds).Find(&medias).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
	}
	mediaMap := make(map[int]live_schedule.MediaProcess)
	for _, v := range medias {
		v := v
		mediaMap[v.Id] = v
	}

	for _, v := range tasks {
		v := v
		//如果回调任务正常并且是审核通过状态
		if now+10*60 > v.AuditEndTime && v.IsAudit == 1 && v.AuditEndTime > 0 {
			e.Log.Trace("将要过期的审核任务", v.AuditTaskId)
			auditId, endTime, err := auditTaskService.CreateAuditTask(v.AuditPullUrl, mediaMap[v.MediaId], baiduService)
			if err != nil {
				e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
			}
			param := make(map[string]interface{})
			param["audit_task_id"] = auditId
			param["audit_end_time"] = endTime
			err = e.Orm.Model(&live_schedule.AuditTask{}).Where("id=?", v.Id).Updates(param).Error
			if err != nil {
				e.Log.Errorf("update AuditTask error:%s \r\n", err)
				return
			}
		}

	}
}

func (e *AuditTaskService) CalculateProgress(partNum int, workTotalNum int, basic live_schedule.ChannelBasic) error {
	result := float64(partNum) / float64(workTotalNum)
	formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
	e.Log.Trace("当前进度为", formattedResult)
	param := make(map[string]interface{})
	param["progress"] = formattedResult
	if workTotalNum == partNum {
		param["channel_state"] = basic.ChannelState
	}
	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Updates(&param).Error
	if err != nil {
		e.Log.Trace("update ChannelBasic error：", err)
	}
	return err
}
